package com.boot.study.table

import com.boot.study.api.SensorReading
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._

/**
 * """
 * root
 * |-- id: STRING
 * |-- temperature: DOUBLE
 *
 * **************************************************
 * root
 * |-- id: STRING
 * |-- temperature: DOUBLE
 *
 * result> (Sensor_1,7.2)
 * result> (Sensor_1,35.8)
 * result> (Sensor_1,31.8)
 * result> (Sensor_1,15.2)
 * result sql> (Sensor_1,35.8)
 * result sql> (Sensor_1,7.2)
 * result sql> (Sensor_1,31.8)
 * result sql> (Sensor_1,15.2)
 * """
 */

object Example {

  def main(args: Array[String]): Unit = {
    // 创建批处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val inputPath: String = "D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt"
    val inputSteam: DataStream[String] = env.readTextFile(inputPath)

    val dataStream: DataStream[SensorReading] = inputSteam.map(data => {
      val arr: Array[String] = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    // 首先创建表执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)
    // 基于流创建一张表
    val dataTable: Table = tableEnv.fromDataStream(dataStream)
    // 调用table api转换
    val resultTable = dataTable
      .select("id, temperature")
      .filter("id == 'Sensor_1'")

    // 输出数据
    resultTable.printSchema()
    resultTable.toAppendStream[(String, Double)].print("result") setParallelism (1)

    println("*" * 50)

    // 直接sql实现
    tableEnv.createTemporaryView("dataTable", dataTable)
    val sql: String = "select id, temperature from dataTable where id = 'Sensor_1'"
    val resultSqlTable = tableEnv.sqlQuery(sql)
    resultSqlTable.printSchema()
    resultSqlTable.toAppendStream[(String, Double)].print("result sql").setParallelism(1)

    env.execute("table api example")
  }
}
